import re
import pandas as pd
import logging
from pandas import DataFrame

from Determining_ad.src.Tools.custom_assert import CustomAssert
from Determining_ad.test.testcases.conftest import custom_error

"""
封装各种判断业务是否正确的方法
"""


class TestingUtils:

    @staticmethod
    def apl_result(custom_error, all_df, ad_df, ad_id, entity, state):

        """
        验证apl埋点的key和value是否存在，是否正确、apl情况是否正确
        :param custom_error:
        :param ad_id:哪一执行的apl操作
        :param all_df:
        :param entity:
        :param state: 共四种情况。1=点击退出弹窗、2=onhide触发apl、3=hdie触发apl上限、4= 点击退出弹窗触发apl上限
        :param ad_df:
        """
        # 获取满足条件的索引
        indices = ad_df[(ad_df['adId'] == ad_id) & (ad_df['adType'] == '18')].index
        if not indices.empty:
            start_index = indices[0]  # 假设只有一个满足条件的索引
            # 使用位置切片获取后续数据
            tem_df = all_df.iloc[start_index:]
            # logging.info(f"tmpdf:{tem_df.to_string()}")
        else:
            custom_error.custom_assert(False, failure_message=f"没有广告{ad_id}的请求事件")
            return False
        pattern = r'key\[(.*?)\],\s+value\[(.*?)\]'

        # 找到bj   在循环中，使用contains匹配Message列中的内容，并提取key和value
        if entity == 'bj':
            mask = tem_df['Message'].str.contains('sdk2_ad_onshow_refresh', na=False)
            mask_df = tem_df[mask]
            if mask_df.empty:
                custom_error.custom_assert(False, failure_message="没有上报apl埋点")
            else:
                # 遍历log信息，找到埋点
                for index, row in mask_df.iterrows():
                    match = re.search(pattern, row['Message'])
                    # logging.info(f"match:{match}")
                    key = match.group(1)
                    value = match.group(2)
                    if key == 'sdk2_ad_onshow_refresh':

                        if state == "2":
                            # 第一种情况 hdie触发apl
                            if value == "2":
                                custom_error.custom_assert(True, success_message="hdie触发apl")
                                break
                            else:
                                custom_error.custom_assert(False, failure_message=f"不是hide触发apl，value={value}")
                                break

                        elif state == "1":
                            #  第二种情况 点击退出弹窗触发apl
                            if value == "1":
                                custom_error.custom_assert(True, success_message="点击110触发apl")
                                break
                            else:
                                custom_error.custom_assert(False, failure_message=f"不是点击110触发apl，value={value}")
                                break
                    elif key == 'sdk2_ad_onshow_refresh|limit':

                        if state == "3":
                            # 第三种情况 hdie触发apl上限
                            if value == "2":
                                custom_error.custom_assert(True, success_message="hide触发apl达上限")
                                break
                            else:
                                custom_error.custom_assert(False, failure_message=f"不是hide触发apl上限，value={value}")
                                break

                        elif state == "4":
                            # 第四种情况  点击退出弹窗触发apl上限
                            if value == "1":
                                custom_error.custom_assert(True, success_message="点击110触发apl上限")
                                break
                            else:
                                custom_error.custom_assert(False, failure_message=f"不是点击110触发apl上限，value={value}")
                                break
                    else:
                        custom_error.custom_assert(False, failure_message="没有找到apl埋点")

        # 找到hn   all_df中满足条件索引
        elif entity == 'hn':
            # 第二种情况 hdie触发apl
            if state == "2":
                mask = tem_df['Message'].str.contains('nativeRe', na=False)
                mask_df = tem_df[mask]
                if mask_df.empty:
                    custom_error.custom_assert(False, failure_message="没有上报apl埋点")
                else:
                    for index, row in mask_df.iterrows():
                        match = re.search(pattern, row['Message'])
                        key = match.group(1)
                        value = match.group(2)
                        if key == 'nativeRe':
                            if value == "2":
                                custom_error.custom_assert(True, success_message="onhide触发apl")
                                break
                            else:
                                custom_error.custom_assert(False, failure_message=f"不是onhide触发apl，value={value}")
                                break

            # 第一种情况 点击退出弹窗触发apl
            elif state == "1":
                mask = tem_df['Message'].str.contains('nativeRe', na=False)
                mask_df = tem_df[mask]
                if mask_df.empty:
                    custom_error.custom_assert(False, failure_message="没有上报apl埋点")
                else:
                    for index, row in mask_df.iterrows():
                        match = re.search(pattern, row['Message'])
                        key = match.group(1)
                        value = match.group(2)
                        if key == 'nativeRe':
                            if value == "1":
                                custom_error.custom_assert(True, success_message="点击110触发apl")
                                break
                            else:
                                custom_error.custom_assert(False, failure_message=f"不是点击110触发apl，value={value}")
                                break

            # 第三种情况 上限后，hdie触发apl
            elif state == "3":
                mask = tem_df['Message'].str.contains('nativeFinishMit', na=False)
                mask_df = tem_df[mask]
                if mask_df.empty:
                    custom_error.custom_assert(False, failure_message="没有上报apl埋点")
                else:
                    for index, row in mask_df.iterrows():
                        match = re.search(pattern, row['Message'])
                        key = match.group(1)
                        value = match.group(2)
                        if key == 'nativeFinishMit':
                            if value == "2":
                                custom_error.custom_assert(True, success_message="onhide触发apl上限")
                                break
                            else:
                                custom_error.custom_assert(False, failure_message=f"不是onhide触发apl上限，value={value}")
                                break

            # 第四种情况 上限后，点击退出弹窗触发apl
            elif state == "4":
                mask = tem_df['Message'].str.contains('nativeFinishMit', na=False)
                mask_df = tem_df[mask]
                if mask_df.empty:
                    custom_error.custom_assert(False, failure_message="没有上报apl埋点")
                else:
                    for index, row in mask_df.iterrows():
                        match = re.search(pattern, row['Message'])
                        key = match.group(1)
                        value = match.group(2)
                        if key == 'nativeFinishMit':
                            if value == "1":
                                custom_error.custom_assert(True, success_message="点击110触发apl上限")
                                break
                            else:
                                custom_error.custom_assert(False, failure_message=f"不是点击110触发apl上限，value={value}")
                                break

    @staticmethod
    def get_ad_sum(custom_error, expect_sum, num_id, df1, df2):
        """
        判断每屏存在广告位个数
        :param custom_error:
        :param expect_sum: 预期广告位个数
        :param num_id: 第几屏
        :param df1:
        :param df2:
        :return:
        """
        # 合并 df1 和 df2，按 NumId 分组求和
        merged = pd.concat([df1, df2])
        result_ad_num = merged.groupby('NumId').size().get(num_id, 0)
        # logging.info(f"result_ad_num:{result_ad_num}")
        custom_error.custom_assert(result_ad_num == expect_sum,
                                   success_message=f"第{num_id}屏个数：预期{expect_sum}个，实际{result_ad_num}个",
                                   failure_message=f"第{num_id}屏个数：预期{expect_sum}个，实际{result_ad_num}个")

    @staticmethod
    def video_event_true(custom_error, ad_id, video_df):
        """
        判断激励视频是否存在 广告事件不正确 的情况
        :param custom_error:
        :param ad_id: 广告位id
        :param video_df:
        :return:
        """
        new_video_df = video_df[video_df['adId'] == ad_id]
        if new_video_df.empty:
            custom_error.custom_assert(False, failure_message=f"{ad_id}激励视频事件为空")
        else:
            result = (new_video_df['check'] != '广告事件不正确').all()
            custom_error.custom_assert(result, success_message=f"{ad_id}激励视频事件正确", failure_message=f"{ad_id}激励视频事件不正确")

    @staticmethod
    def ad111_event_true(custom_error, num_id, df1, df2, expect_second_sum):
        """
        判断每屏111广告事件是否正常
        :param expect_second_sum: 子广告预期个数
        :param custom_error:
        :param num_id: 第几屏
        :param df1:
        :param df2:
        :return:
        """

        def check_event(target_df, target_num_id):
            if target_df is None:
                return None
            else:
                return target_df[target_df['NumId'] == target_num_id]

        new_df1 = check_event(df1, num_id)
        new_df2 = check_event(df2, num_id)
        if new_df1 is None or new_df1.empty:
            custom_error.custom_assert(False, failure_message=f"主广告位第{num_id}屏事件为空")
        elif new_df2 is None:
            custom_error.custom_assert(False, failure_message=f"子广告位第{num_id}屏事件为空")
        elif new_df2.empty:
            custom_error.custom_assert(expect_second_sum == 0,
                                       success_message=f"第{num_id}屏没有配置子广告位",
                                       failure_message=f"第{num_id}屏子广告个数应为{expect_second_sum}，实际为0")
        else:
            result1 = (new_df1['check'] != '广告事件不正确').all()
            result2 = (new_df2['check'] != '广告事件不正确').all()
            result3 = result1 and result2
            custom_error.custom_assert(result3, success_message=f"第{num_id}屏事件都正确", failure_message=f"第{num_id}屏事件存在不正确")

    @staticmethod
    def ad110_event_true(custom_error, *, quit_df=None, quit_second_df=None):
        """
        判断110退出弹窗事件是否正确
        :param custom_error:
        :param quit_df:
        :param quit_second_df:
        :return:
        """
        if quit_df.empty:
            custom_error.custom_assert(False, failure_message=f"主退出弹窗广告事件为空")
        elif quit_second_df.empty:
            custom_error.custom_assert(False, failure_message=f"子退出弹窗广告事件为空")
        elif not quit_df.empty and not quit_second_df.empty:
            result1 = (quit_df['check'] != '广告事件不正确').all()
            result2 = (quit_second_df['check'] != '广告事件不正确').all()
            result3 = result1 and result2
            custom_error.custom_assert(result3, success_message=f"退出弹窗广告事件正确", failure_message=f"退出弹窗广告事件不正确")

    @staticmethod
    def click_event_result(custom_error, ad_id, *, num_id=None, df_111=None, df_110=None):
        """
        判断111和110点击事件
        :param df_110:
        :param df_111:
        :param custom_error:
        :param ad_id: 广告位id
        :param num_id: 第几屏,111必传
        :return:
        """
        real = "空广告事件"

        # 判断退出弹窗事件
        if df_111 is None:
            expect_110 = (df_110[(df_110['adId'] == ad_id)]['check'] == '点击了广告')
            if df_110.empty:
                custom_error.custom_assert(False, failure_message=f"{ad_id}事件预期值：点击了广告，实际：{real}")
            else:
                real = df_110[(df_110['adId'] == ad_id)]['check'].iloc[0]
                custom_error.custom_assert(expect_110.all(),
                                           failure_message=f"{ad_id}事件预期值：点击了广告，实际：{real}",
                                           success_message=f"{ad_id}事件预期值：点击了广告，实际：{real}",
                                           )
        # 判断111事件
        elif df_110 is None:
            expect_111 = (df_111[(df_111['adId'] == ad_id) & (df_111['NumId'] == num_id)]['check'] == '点击了广告')
            if df_111.empty:
                custom_error.custom_assert(False, failure_message=f"第{num_id}屏的{ad_id}事件预期值：点击了广告，实际：{real}")
            else:
                try:
                    real = df_111[(df_111['adId'] == ad_id) & (df_111['NumId'] == num_id)]['check'].iloc[0]
                except IndexError:
                    custom_error.custom_assert(False, failure_message=f"没有第{num_id}屏的{ad_id}广告")
                else:
                    custom_error.custom_assert(expect_111.all(),
                                               failure_message=f"第{num_id}屏的{ad_id}事件预期值：点击了广告，实际：{real}",
                                               success_message=f"第{num_id}屏的{ad_id}事件预期值：点击了广告，实际：{real}",
                                               )

    @staticmethod
    def img_num_result(custom_error, except_num, img, num_id, df1, df2):
        """
        判断 每屏xxx广告源 的个数是否正确
        :param custom_error:
        :param except_num: 预期个数
        :param img: 广告源
        :param num_id: 第几屏
        :param df1:
        :param df2:
        :return:
        """

        def count_ads(df, target_id, target_img):
            """计算给定DataFrame中符合条件的广告数量"""
            if df is None:
                return 0
            mask = (df['NumId'] == target_id) & (df['img'].str.contains(target_img, na=False))
            return len(df[mask])

        # 计算两个屏幕中的广告数量
        count_first = count_ads(df1, num_id, img)
        count_second = count_ads(df2, num_id, img)

        # 断言预期数量与实际数量是否一致
        custom_error.custom_assert(
            count_first + count_second == except_num,
            success_message=f"预期{except_num}个{img}，实际{count_first + count_second}个{img}",
            failure_message=f"预期{except_num}个{img}，实际{count_first + count_second}个{img}"
        )

    @staticmethod
    def button_event_result(custom_error, ad_id, num_id, *, df_111=None, df_110=None):
        """
        判断某个广告位是否有 button事件
        :param df_110:
        :param df_111:
        :param num_id:
        :param custom_error:
        :param ad_id: 广告位id
        :return:
        """
        if df_111 is None:
            # 110的button判断
            expect = (df_110[(df_110['adId'] == ad_id)]['check'] == '点击了adbutton')
            real = "空广告事件"
            if expect.empty:
                # 空的Series.all()返回True
                custom_error.custom_assert(~expect.all(), failure_message=f"{ad_id}事件预期值：'点击了adbutton'，实际：{real}")
            else:
                real = df_110[(df_110['adId'] == ad_id)]['check'].iloc[0]
                custom_error.custom_assert(expect.all(),
                                           success_message=f"{ad_id}预期事件为：'点击了adbutton'，实际：{real}",
                                           failure_message=f"{ad_id}预期事件为：'点击了adbutton'，实际：{real}")
        else:
            # 111的button判断
            expect = (df_111[(df_111['adId'] == ad_id) & (df_111['NumId'] == num_id)]['check'] == '点击了adbutton')
            real = "空广告事件"
            if expect.empty:
                # 空的Series.all()返回True
                custom_error.custom_assert(~expect.all(), failure_message=f"第{num_id}屏的{ad_id}事件预期值：'点击了adbutton'，实际：{real}")
            else:
                real = df_111[(df_111['adId'] == ad_id) & (df_111['NumId'] == num_id)]['check'].iloc[0]
                custom_error.custom_assert(expect.all(),
                                           success_message=f"第{num_id}屏的{ad_id}预期事件为：'点击了adbutton'，实际：{real}",
                                           failure_message=f"第{num_id}屏的{ad_id}预期事件为：'点击了adbutton'，实际：{real}")

    @staticmethod
    def video_num_result(custom_error, video_df, expect_num, video_id):
        """
        判断激励视频个数是否正确
        验证逻辑：展示成功等个数符合 且 之后不会再有请求，则通过
        如果竞价失败，则直接断言失败；

        目前根据用例，只兼容个数为1的情况
        :param custom_error:
        :param video_df:
        :param expect_num: 预期个数
        :param video_id: 广告位id
        :return:
        """
        video_df = video_df[video_df['adId'] == video_id]
        check_counts = video_df['check'].value_counts()
        # 找到目标广告事件的个数
        target_checks = ["展示成功", "提前关闭", "播放完毕", "后台无法播放"]
        result_ad_num = check_counts[check_counts.index.isin(target_checks)].sum()

        # if判断 如果check值不符合预期，直接断言失败
        # 如果都是竞价失败，也当做断言失败处理
        if result_ad_num == expect_num:
            for check_value in target_checks:  # 直接遍历值，无需索引
                # 符合目标的广告事件，组成一个df
                matched_rows = video_df[video_df['check'] == check_value]

                # 跳过未找到的行
                if matched_rows.empty:
                    if expect_num == 0:
                        custom_error.custom_assert(True, success_message=f"激励视频{video_id}期望{expect_num}个，实际{result_ad_num}个")
                        break
                    continue

                # 遍历所有匹配行（即使有多个）
                for idx in matched_rows.index:
                    # 提取当前行之后的数据（包含当前行的下一行）
                    sliced_data = video_df.iloc[idx + 1:]
                    if not sliced_data.empty:
                        custom_error.custom_assert(False, failure_message=f"激励视频{video_id}期望成功{expect_num}个，实际{result_ad_num}个")
                        break
                custom_error.custom_assert(True, success_message=f"激励视频{video_id}期望成功{expect_num}个，实际{result_ad_num}个")
        else:
            custom_error.custom_assert(False, failure_message=f"激励视频{video_id}期望成功{expect_num}个，实际{result_ad_num}个")

    @staticmethod
    def no_first_ad(custom_error, first_df, ad_df, all_df, NumId, img, adId):
        """
        验证json配置配置中，非nofirst的情况
        只有1111竞价失败时，才会上报
        :param custom_error:
        :param first_df:
        :param ad_df:
        :param all_df:
        :param NumId:
        :param img:
        :param adId:
        :return:
        """

        bj_value = 'sdk2_sdk_oninit_noFirst'

        # 筛查该广告位为展示img广告源的情况
        mask_first = (first_df['NumId'] == NumId) & (first_df['img'].str.contains(img, na=False))

        if mask_first.empty:
            custom_error.custom_assert(False, failure_message=f"没有相应的广告")
        else:
            if mask_first.iloc[0]:
                # 不符合预期 展示了某个源的广告
                # 展示了非目标广告源的广告，是不会报那个埋点的
                custom_error.custom_assert(False, failure_message=f"{adId}不符合预期，展示了{img}")
            else:
                # 判断该广告位展示的广告源是否为预期值
                real_event = first_df.loc[first_df["NumId"] == NumId, "check"][0]
                real_img = first_df.loc[first_df["NumId"] == NumId, "img"][0]
                custom_error.custom_assert(~mask_first.iloc[0],
                                           success_message=f"{adId}没有展示{img}广告，实际为：{adId}{real_event}了{real_img}的广告",
                                           failure_message=f"{adId}预期没有{img}广告，实际存在{img}")
                if real_img == img:
                    # 找到符合 目标广告位失败的索引，一般失败后5条日志内，会有相应的埋点上报
                    indices3 = ad_df[(ad_df['adId'] == adId) & (ad_df['adType'] == '5')].index.tolist()[0]
                    indices4 = indices3 + 5
                    all_df = all_df.loc[indices3:indices4]
                    mask_df = all_df[all_df['Message'].str.contains(bj_value, na=False)]

                    if mask_df.empty:
                        custom_error.custom_assert(False, failure_message=f"{adId}没有上报{bj_value}")
                    else:
                        custom_error.custom_assert(True, success_message=f"{adId}上报了{bj_value}")

    # 这个埋点好像现在不报了
    # @staticmethod
    # def first_ad(custom_error, first_df, ad_df, all_df, NumId, img, adId):

    # bj_value='sdk2_sdk_oninit_isFirst'
    #
    #
    # # sdk2_sdk_oninit_isFirst埋点是否存在
    # indices1 = self.ad_df[(self.ad_df['adId'] == '402') & (self.ad_df['adType'] == '18')].index.todo_list()[0]
    # logging.info(f"indices1:{indices1}")
    # indices2 = self.ad_df[(self.ad_df['adId'] == '402') & (self.ad_df['adType'] == '6')].index.todo_list()[0]
    # logging.info(f"indices2:{indices2}")
    # all_df = self.all_df.loc[indices1:indices2]
    # logging.info(f"索引all_df:{all_df}")
    # if self.entity == 'bj':
    #     mask_df = all_df[all_df['Message'].str.contains('sdk2_sdk_oninit_isFirst', na=False)]
    #     logging.info(f"主广告位展示mask_df:{mask_df}")
    #     if mask_df.empty:
    #         custom_error.custom_assert(False, "402的18和6之间上报sdk2_sdk_oninit_isFirst", "402的18和6之间没有上报sdk2_sdk_oninit_isFirst")

    @staticmethod
    def bqt_num(custom_error, ad_df):
        """
        判断百度请求个数能否被限制
        :param custom_error:
        :param ad_df:
        :return:
        """
        # 百度请求成功个数不能超过3个，但是有缓存，一般不超过6个，我多写几个
        ad_df = ad_df[(ad_df['img'] == 'other-bqt') & (ad_df['adType'] == '6')]
        num = 8
        if ad_df.shape[0] == 0:
            custom_error.custom_assert(False, failure_message=f"百度请求全部失败，个数限制预期{num}个，实际{ad_df.shape[0]}个")
        else:
            custom_error.custom_assert(num > ad_df.shape[0],
                                       success_message=f"百度个数限制预期{num}个，实际{ad_df.shape[0]}个",
                                       failure_message=f"百度个数限制预期{num}个，实际{ad_df.shape[0]}个")

    @staticmethod
    def check_key(custom_error, all_df, entity):
        """
        判断重要埋点是否上报
        :param custom_error:
        :param entity:
        :param all_df:
        :return:
        """
        # 只有存在这个标记时，才执行下面的代码
        # 首先要在pytest.ini中标记
        # marker = request.node.get_closest_marker("check_key_setup")
        # if marker:
        #     test_instance = request.instance
        #     entity = getattr(test_instance, 'entity', None)
        #     all_df = getattr(test_instance, 'all_df', None)

        if entity == 'bj':
            # 正常操作预期上报
            key_list = ['sdk2_io_dp', 'sdk2_appux_hide', 'sdk2_appux_show', 'sdk2_single_onrefresh_', 'sdk2_dp_first_', 'quick_deeplink', 'sdk2_adconfig_req_source',
                        'sdk2_adfireq_init', 'sdk2_adfireq_sc',
                        'sdk2_ua_init', 'sdk2_source_init', 'sling_sole_id', 'sdk2_sw_pop_', 'user2_ck_pop_', 'sdk2_videopool_onplay_start', 'sdk2_priv_init',
                        'sdk2_priv_ag2']
            # 正常操作预期不上报
            nonexistent_key_list = ['sdk2_priv_sw', 'sdk2_adreq_tio', 'sdk2_adfireq_tio', 'performance']
        elif entity == 'hn':
            key_list = ['sdk2_io_dp', 'appHide', 'appShow', 'sdk2_single_onrefresh_', 'indexFirst', 'quick_deeplink', 'adConfigReqOrigin',
                        'adInit', 'adFWin',
                        'uaInit', 'originInit', 'sling_sole_id', 'popShow', 'popClick', 'videoPlayStart', 'yinsiInit',
                        'yinsiAgree2']
            nonexistent_key_list = ['yinsiShow', 'adReqFTiO', 'adReqTiO', 'performance']
        else:
            custom_error.custom_assert(False, failure_message=f"主体不存在：{entity}")
            return

        custom_events = all_df[all_df["Message"].str.contains(r"=+\s*自定义事件\s*=+", na=False)].copy()
        # 提取 key 的值（例如 ADSDK、OTHER_KEY）
        custom_events["key"] = custom_events["Message"].str.extract(r"key$$([^$$]+)\]")
        custom_events["key"] = custom_events["Message"].str.extract(r"key\[([^\]]+)\]")
        # logging.info(f"custom_events  ：\n{custom_events['key'].to_string()}")
        existing_keys = custom_events["key"].dropna().unique().tolist()

        # 判断该存在的埋点
        missing_keys = list(set(key_list) - set(existing_keys))
        # logging.info(f"已上报的 key ：{existing_keys}")
        if missing_keys:
            custom_error.custom_assert(False, failure_message=f"以下 key 预期上报：{missing_keys}")
        else:
            custom_error.custom_assert(True, success_message=f" key 均已上报")

        # 判断不该存在的埋点
        nonexistent_missing_keys = list(set(nonexistent_key_list) & set(existing_keys))
        if nonexistent_missing_keys:
            custom_error.custom_assert(False, failure_message=f"以下 key 预期不上报：{nonexistent_missing_keys}")
        else:
            custom_error.custom_assert(True, success_message=f" key 均未上报")
